package com.zhou.common.mybatis.template.oss.mq.consumer.local;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zhou.common.base.mq.mq.local.RocketMqProperties;
import com.zhou.common.mybatis.template.oss.model.OssUploadResult;
import com.zhou.common.mybatis.template.oss.mq.consumer.MqConsumerProperties;
import com.zhou.common.mybatis.template.oss.service.OssService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.nio.charset.StandardCharsets;

@Slf4j
@Configuration
@ConditionalOnProperty(prefix = "rocketmq", name = "enable", havingValue = "true")
public class MqConfig {

    @Autowired
    private RocketMqProperties rocketMqProperties;

    @Autowired
    private MqConsumerProperties mqConsumerProperties;

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private OssService ossService;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    @ConditionalOnProperty(prefix = "rocketmq.oss.consumer", name = "enable", havingValue = "true")
    public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mqConsumerProperties.getGroupId());
        consumer.setNamesrvAddr(rocketMqProperties.getNameServer());
        consumer.subscribe(mqConsumerProperties.getTopic(), mqConsumerProperties.getTag());
        /**
         * 设置consumer第一次启动是从队列头部开始还是队列尾部开始
         * 如果不是第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {

            try {
                for (Message msg : msgs) {

                    String body = new String(msg.getBody(), StandardCharsets.UTF_8);
                    log.info("receive body={}", body);
                    OssUploadResult ossUploadResult = objectMapper.readValue(body, new TypeReference<OssUploadResult>() {});
                    ossService.service(ossUploadResult);
                }
            } catch (Exception e) {
                log.error("", e);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        return consumer;
    }
}


